
package cfar.parallel.reduce;

import cfar.ff.data.Assocs;
import cfar.ff.data.Assocs.Assoc2;
import cfar.ff.data.Assocs.Assoc3;
import cfar.ff.data.List;
import cfar.ff.data.Maybe;
import cfar.ff.fun.F1;
import cfar.ff.fun.F2;
import cfar.ff.fun.F3;
import cfar.ff.fun.F4;
import cfar.ff.fun.Seq;
import cfar.ff.interfaces.Sequence;

import static cfar.ff.data.Maybe.*;
import static cfar.ff.fun.Curry.*;

/**
 *
 * @author badi
 */
public class RReduce<S,A> implements F4< Sequence<S,Integer>
                                       , F1<Sequence<S,A>,A>
                                       , F2<A,A,A>
                                       , Sequence<S,A>
                                       , A
                                       >
{

    /***************************************************************/
    public A f( final Sequence<S, Integer>        redundancies
              , final F1<Sequence<S, A>, A>       chooser
              , final F2<A, A, A>                 f
              , final Sequence<S, A>              seq
              )
    {
        F1<Assoc3<Integer,A,A>, Assoc2<Integer,A>> apply = new F1<Assoc3<Integer, A, A>, Assoc2<Integer,A>>() {

            public Assoc2<Integer,A> f(Assoc3<Integer, A, A> a) {
                Sequence<S,Assoc2<A,A>> s = (Sequence<S, Assoc2<A, A>>) seq.<A>empty();
                for(int i = 0; i < a.$1; i++)
                    s = s.cons(new Assoc2<A, A>(a.$2, a.$3));

                F1<Assoc2<A,A>, Assoc2<Integer,A>> apply_to_replications = curry(curry(rreduce, f), a.$1);
                
                Sequence<S,Assoc2<Integer,A>> s2 = Seq.parMap(apply_to_replications, s);
                Sequence<S,A> s3 = (Sequence<S, A>) Seq.parMap(Assocs.<Integer, A>snd(), s2);

                return new Assoc2<Integer, A>(a.$1, chooser.f(s3));
            }
        };

        Sequence<S, Assoc2<Integer, A>> zipped = Seq.zip(redundancies, seq);
        do {
            Assoc2<List<Assoc3<Integer,A, A>>, Maybe<Assoc2<Integer,A>>> assoc = mkAssocs(zipped);
            Sequence<S, Assoc2<Integer,A>>
                    partially_reduced =
                    (Sequence<S, Assoc2<Integer, A>>) Seq.parMap(apply, assoc.$1);

            zipped = assoc.$2.isJust
                    ? partially_reduced.snoc(assoc.$2.fromJust())
                    : partially_reduced
                    ;
        } while (Seq.length(zipped) != 1);

        return Seq.head(zipped).$2;
    }
    /***************************************************************/

    private F3<F2<A,A,A>, Integer, Assoc2<A,A>, Assoc2<Integer,A>> rreduce = new F3<F2<A, A, A>, Integer, Assoc2<A, A>, Assoc2<Integer,A>>() {

        public Assoc2<Integer, A> f(F2<A, A, A> f, Integer r, Assoc2<A, A> a) {
            return new Assoc2<Integer, A>(r, f.f(a.$1, a.$2));
        }
    };

    /**
     *
     * @param <A>
     * @param redundancies
     * @param seq
     * @return a 3-tuple of (replication, first, second) where replication = max(rep_first, rep_second)
     */
    private Assoc2<List<Assoc3<Integer,A, A>>, Maybe<Assoc2<Integer,A>>> mkAssocs( Sequence<S,Assoc2<Integer,A>> vals)
    {
        List<Assoc3<Integer, A, A>> l = List.nil();

        int need = 2;

        Maybe<Assoc2<Integer,A>> tmp = nothing();

        for(Assoc2<Integer,A> assoc : vals) {
            if (need == 2) {
                tmp = just(assoc);
                need--;
                continue;
            } else if (need == 1) {
                int r = tmp.fromJust().$1 < assoc.$1
                        ? assoc.$1
                        : tmp.fromJust().$1
                        ;
                
                l = l.snoc(new Assoc3<Integer,A, A>(r, tmp.fromJust().$2, assoc.$2));
                need = 2;
                tmp = nothing();
                continue;
            } else {
                throw new RuntimeException("The impossible occured!");
            }
        }

        // if seq had an odd number elements tmp should be 'just'
        return new Assoc2<List<Assoc3<Integer, A, A>>, Maybe<Assoc2<Integer, A>>>(l, tmp);
    }


}
